package io.github.melin.sqlflow.metadata;

import io.github.melin.sqlflow.tree.QualifiedName;

import java.util.*;

/**
 * huaixin 2021/12/25 6:13 PM
 */
public class SimpleMetadataService implements MetadataService {

    private final List<SchemaTable> tables = new ArrayList<>();

    private final Optional<String> defaultSchema;

    private final Optional<String> defaultCatalog;

    public SimpleMetadataService(String schema) {
        this.defaultCatalog = Optional.empty();
        this.defaultSchema = Optional.of(schema);
    }

    public SimpleMetadataService(String defaultCatalog, String defaultSchema) {
        this.defaultCatalog = Optional.of(defaultCatalog);
        this.defaultSchema = Optional.of(defaultSchema);
    }

    public MetadataService addTableMetadata(List<SchemaTable> schemaTables) {
        for (SchemaTable schemaTable : schemaTables) {
            addTableMetadata(schemaTable);
        }
        return this;
    }

    public MetadataService addTableMetadata(SchemaTable schemaTable) {
        String schemaTableKey = schemaTable.toString();
        Optional<SchemaTable> existingTable =  tables.stream()
                .filter(t -> t.toString().equalsIgnoreCase(schemaTableKey))
                .findFirst();

        if (existingTable.isPresent()) {
            // 如果存在，合并列
            List<String> mergedColumns = mergeColumns(existingTable.get().getColumns(), schemaTable.getColumns());
            existingTable.get().setColumns(mergedColumns);
            // 如果分区存在，合并分区列
            List<String> mergedpartitionColumns = mergeColumns(existingTable.get().getPartitionColumns(), schemaTable.getPartitionColumns());
            existingTable.get().setPartitionColumns(mergedpartitionColumns);
        } else {
            // 如果不存在，添加新的 SchemaTable
            tables.add(schemaTable);
        }
        return this;
    }


    public  List<SchemaTable> getTables() {
        return tables;
    }

    @Override
    public Optional<String> getSchema() {
        return defaultSchema;
    }

    @Override
    public Optional<String> getCatalog() {
        return defaultCatalog;
    }

    @Override
    public boolean isAggregationFunction(QualifiedName name) {
        return false;
    }

    @Override
    public Optional<SchemaTable> getTableSchema(QualifiedObjectName table) {
        return tables.stream()
                .filter(t -> t.toString().equalsIgnoreCase(table.toString()))
                .findFirst();
    }

    @Override
    public Optional<ViewDefinition> getView(QualifiedObjectName viewName) {
        return Optional.empty();
    }

    private List<String> mergeColumns(List<String> existingColumns, List<String> newColumns) {
        if(existingColumns==null){
            existingColumns=new ArrayList<>();
        }
        if(newColumns==null){
            newColumns=new ArrayList<>();
        }
        // 合并两个列列表并去重
        Set<String> mergedColumnsSet = new HashSet<>(existingColumns);
        mergedColumnsSet.addAll(newColumns);
        return new ArrayList<>(mergedColumnsSet);
    }
}